//This file is a part of StoragePlug for Minecraft.
//http://code.google.com/p/minecraft-storageplug/
//StoragePlug is Copyright (C) 2011 Christopher Smith
//Released under the MIT license:
//
//Permission is hereby granted, free of charge, to any person obtaining a copy
//of this software and associated documentation files (the "Software"), to deal
//in the Software without restriction, including without limitation the rights 
//to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//copies of the Software, and to permit persons to whom the Software is
//furnished to do so, subject to the following conditions:
//
//The above copyright notice and this permission notice shall be included in
//all copies or substantial portions of the Software.
//
//THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
//FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
//DEALINGS IN THE SOFTWARE.
//
//Related to Minecraft Copyright (C) 2009-2011 Mojang http://www.minecraft.net/

package net.minecraft.storageplug;

import java.util.AbstractMap.SimpleEntry;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

public class WriteBatcher 
    implements StorageProvider
{
    final StorageProvider next;
    volatile CountDownLatch engageBatchThread;
    volatile CountDownLatch batchCompleteNotify; //always has a value, 1 when a batch is in progress
    volatile CountDownLatch batchThreadTerminated;
    volatile boolean terminateBatchThread;
    
    //current modified items, not yet being written
    //protected by lock
    volatile HashMap<Key, DataTree> modified; 
    
    //items being written, may not be in next provider yet
    //not directly mutated: replaced with `modified` by worker under lock
    volatile HashMap<Key, DataTree> pending; 
    
    public WriteBatcher(StorageProvider next)
    {
        this.next = next;
        
        engageBatchThread = new CountDownLatch(1); //no work for thread at start
        batchCompleteNotify = new CountDownLatch(0); //start as though the last batch completed already
        batchThreadTerminated = new CountDownLatch(1); //thread not terminated at start
        terminateBatchThread = false; //don't kill the thread until dispose
        
        pending = new HashMap<Key, DataTree>();
        modified = new HashMap<Key, DataTree>();
        
        Thread workerThread = new Thread(new Worker(), "SPlug Writer");
        workerThread.start();
    }
        
    private boolean synchroIsCacheEmpty()
    {
        synchronized(this)
        {
            return modified.isEmpty() && pending.isEmpty();
        }
    }
    
    class Worker
        implements Runnable
    {
        @Override
        public void run()
        {
            //only exit when the thread is asked to term AND the cache is drained
            while(!terminateBatchThread || !synchroIsCacheEmpty())
            {
                if(synchroIsCacheEmpty()) //in case we missed items being added while the last batch was running
                    try
                    {
                        engageBatchThread.await();
                    } catch (InterruptedException ex)
                    {
                        //this should never happen
                        ex.printStackTrace();
                    }

                //capture modified items
                synchronized(this)
                {
                    pending = modified;
                    modified = new HashMap<Key, DataTree>();
                }

                ArrayList<SimpleEntry<Key, DataTree>> batch = new ArrayList<SimpleEntry<Key, DataTree>>(pending.size());

                for(Map.Entry<Key, DataTree> e : pending.entrySet())
                    batch.add(new SimpleEntry<Key, DataTree>(e.getKey(), e.getValue()));

                try
                {
                    System.out.print("W" + batch.size());
                    next.writeDataTreeBatch(batch);
                    System.out.print("/");
                }
                catch (Exception ex)
                {
                    ex.printStackTrace();
                }

                //empty pending
                pending = new HashMap<Key, DataTree>();

                engageBatchThread = new CountDownLatch(1);
            }
            batchThreadTerminated.countDown();
        }
    }

    @Override
    public void writeDataTree(DataTree dt, Key key)
        throws Exception
    {
        synchronized(this)
        {
            modified.put(key, dt);
            engageBatchThread.countDown(); //signal the batch thread that it can start            
        }
    }
    
    @Override
    public void writeDataTreeBatch(Collection<? extends Map.Entry<Key, DataTree>> list)
        throws Exception
    {
        for(Map.Entry<Key, DataTree> e : list)
            writeDataTree(e.getValue(), e.getKey());
    }

    @Override
    public DataTree readDataTree(Key key)
        throws Exception
    {
        synchronized(this)
        {
            DataTree dt = modified.get(key);
            if(dt != null)
                return dt;

            dt = pending.get(key);
            if(dt != null)
                return dt;
        }
                
        return next.readDataTree(key);
    }

    @Override
    public Collection<? extends Map.Entry<Key, DataTree>> readDataTreeBatch(Collection<? extends Key> list)
        throws Exception
    {
        ArrayList<Map.Entry<Key, DataTree>> out = new ArrayList<Map.Entry<Key, DataTree>>();
        ArrayList<Key> remainingKeys = new ArrayList<Key>(list);

        //take care of the ones that are pending to write first
        synchronized(this)
        {
            for(Key key : list)
            {
                DataTree dt = modified.get(key);
                if(dt != null)
                {
                    out.add(new SimpleEntry<Key, DataTree>(key, dt));
                    remainingKeys.remove(key);
                }
                else
                {
                    dt = pending.get(key);
                    if(dt != null)
                    {
                        out.add(new SimpleEntry<Key, DataTree>(key, dt));
                        remainingKeys.remove(key);
                    }
                }                
            }
        }
        
        //if there's anything left, pass the request on
        if(!remainingKeys.isEmpty())
            out.addAll(next.readDataTreeBatch(remainingKeys));
        
        return out;
    }

    @Override
    public void writeUniqueIDs(Collection<? extends Map.Entry<String, Integer>> map)
        throws Exception
    {
        next.writeUniqueIDs(map);
    }

    @Override
    public Collection<? extends Map.Entry<String, Integer>> readUniqueIDs()
        throws Exception
    {
        return next.readUniqueIDs();
    }

    @Override
    public int nextUniqueID(String key)
        throws Exception
    {
        return next.nextUniqueID(key);
    }

    @Override
    public Collection<? extends Key> iterKeys(Class type)
        throws Exception
    {
        ArrayList<Key> out = new ArrayList<Key>();
        for(Key k : next.iterKeys(type))
            out.add(k);
        
        synchronized(this)
        {
            for(Key k : modified.keySet())
                if(k.getClass().equals(type))
                    out.add(k);

            for(Key k : pending.keySet())
                if(k.getClass().equals(type))
                    out.add(k);
        }
        
        return out;
    }

    @Override
    public void flush()
        throws Exception
    {
        //the cache only has modified data in it
        //wait for the cache to be emptied
        //usually, it's not empty because a batch is in progress
        // so wait for a batch to complete before trying again
        
        //there is a possibility that the cache has data in it but
        // the worker has not yet started; this will cause a temporary
        // busy wait looping here until the batch starts,
        // so that this thread can sleep until completion
        while(!synchroIsCacheEmpty())
            batchCompleteNotify.await();
    }

    @Override
    public void dispose()
        throws Exception
    {
        terminateBatchThread = true;
        engageBatchThread.countDown();
        batchThreadTerminated.await();
    }
}
